package xin.yangshuai.zookeeper01.case2;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class DistributedLock {

    // 注意：逗号左右不能有空格
    private String connectString = "192.168.145.132:2181,192.168.145.133:2181,192.168.145.134:2181";
    // 2000毫秒
    private int sessionTimeout = 2000;

    private ZooKeeper zkClient;

    private CountDownLatch countDownLatch = new CountDownLatch(1);

    private CountDownLatch waitLatch = new CountDownLatch(1);

    private String currentMode;
    private String listeningNode;

    public DistributedLock() throws IOException, InterruptedException, KeeperException {

        // 获取连接
        zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                // 初始化时，会执行一次
                System.out.println(watchedEvent.getPath() + "----------------------------");

                // 启动时会触发，表示已经连接上zookeeper了，放行
                if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
                    countDownLatch.countDown();
                }

                // 触发监听，监听的节点被删除了，放行
                if (watchedEvent.getType() == Event.EventType.NodeDeleted && watchedEvent.getPath().equals(listeningNode)) {
                    waitLatch.countDown();
                }
            }
        });

        countDownLatch.await();
        // 已经连接上zookeeper后，继续执行

        // 判断父节点/locks是否存在
        Stat stat = zkClient.exists("/locks", false);
        if (stat == null) {
            // 创建父节点/locks
            zkClient.create("/locks", "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    // 获取锁
    public void lock() throws KeeperException, InterruptedException {

        // 创建临时有序号节点
        currentMode = zkClient.create("/locks/seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

        Thread.sleep(10);

        // 获取/locks节点的子节点列表
        List<String> children = zkClient.getChildren("/locks", false);

        if (children.size() == 1) {
            // 只有一个节点，直接获取锁，返回
            return;
        }

        Collections.sort(children);

        // 获取节点名称，seq-00000000
        String thisNode = currentMode.substring("/locks/".length());
        // 获取节点在集合中的位置
        int index = children.indexOf(thisNode);

        if (index == 0) {
            // 是第一个节点，直接获取锁，返回
            return;
        }

        // 获取前一个索引的节点
        listeningNode = "/locks/" + children.get(index - 1);
        // 监听前一个节点的值的变化
        zkClient.getData(listeningNode, true, null);

        // 线程等待，直到上一个节点释放锁（被删除）时，将触发监听回调，会设置此处放行
        System.out.println(Thread.currentThread().getName() + " 未获取到锁，等待");
        waitLatch.await();
        // 上一个节点释放锁，继续执行，表示获取到锁
    }

    // 解锁
    public void unlock() throws KeeperException, InterruptedException {

        // 删除节点，表示释放锁
        zkClient.delete(currentMode, -1);
    }
}
